package com.vcg.community.filter;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.amazonaws.services.dynamodbv2.model.AttributeValue;
import com.amazonaws.services.dynamodbv2.model.StreamRecord;
import com.amazonaws.services.dynamodbv2.streamsadapter.model.RecordAdapter;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessor;
import com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorCheckpointer;
import com.amazonaws.services.kinesis.clientlibrary.types.ShutdownReason;
import com.amazonaws.services.kinesis.model.Record;
import com.fivepx.utils.MapUtils;

import java.util.List;
import java.util.Map;

/**
 * 表中流数据分片处理器
 * @author felayman
 * @since 16/1/5
 */

public class StreamsRecordProcessor implements IRecordProcessor {

    private Integer checkpointCounter;

    public void initialize(String shardId) {
        checkpointCounter = 0;
    }

    public void processRecords(List<Record> records,IRecordProcessorCheckpointer checkpointer) {
        for(Record record : records) {
            if (record instanceof RecordAdapter) {
                com.amazonaws.services.dynamodbv2.model.Record recordAdapter = ((RecordAdapter) record).getInternalObject();
                StreamRecord streamRecord = recordAdapter.getDynamodb();
                String eventName = recordAdapter.getEventName();
                DataSwitcher dataSwitcher = new DataSwitcher();
                if (eventName.equals("REMOVE")) {
                    Map<String, AttributeValue> oldMap = streamRecord.getOldImage();
                    Map<String, Object> dataMap = MapUtils.toSingleMap(oldMap);
                    JSONObject result = JSON.parseObject(JSON.toJSONString(dataMap));
                    dataSwitcher.transform(eventName, result.toJSONString());
                } else {
                    Map<String, AttributeValue> stringAttributeValueMap = streamRecord.getNewImage();
                    Map<String, Object> dataMap = MapUtils.toSingleMap(stringAttributeValueMap);
                    JSONObject result = JSON.parseObject(JSON.toJSONString(dataMap));
                    dataSwitcher.transform(eventName, result.toJSONString());
                }
            }
            try {
                checkpointer.checkpoint();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    public void shutdown(IRecordProcessorCheckpointer checkpointer,ShutdownReason reason) {
        if(reason == ShutdownReason.TERMINATE) {
            try {
                checkpointer.checkpoint();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

}